package com.atguigu.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TableProcess;
import com.atguigu.func.DimSinkFunction;
import com.atguigu.func.DimTableProcessFuntion;
import com.atguigu.utils.KafkaUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;


// 数据流：web/app->mysql(binlog)->maxwell->kafka(ODS)->FlinkKafkaConsumer（主流）
//                                                                     ==>phoenix(DIM)
//                 mysql（binlog）->flinkCDC(配置流)
// 程 序: Mock   ->mysql(binlog)->maxwell->kafka(ZK) ->DimApp(hdfs、zk、hbase)->phoenix(DIM)
public class DimApp {

    public static void main(String[] args) throws Exception {
        //todo 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);//生产环境中设置为kafka主题的分区数（并行消费）

        //todo 生产环境一定要写，测试注释掉，否则每次测试都得开hdfs
//        需要从checkpoint或savepoint启动程序
//        //2.1 开启checkpoint，每隔5s钟做一次ck，并指定ck的一致性语义
//        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);//exactly once：默认barrier对齐
//        //2.2 设置超时时间为1min
//        env.getCheckpointConfig().setCheckpointTimeout(60*1000L);//设置超时时间设置checkpoint的超时时间为1min，是指做一次checkpoint的时间；如果超时则认为本次checkpoint失败，这个checkpoint就丢了，继续一下一次checkpoint即可
//        //2.3设置两次重启的最小时间间隔为3s
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        //2.4设置任务关闭的时候保留最后一次ck数据
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        //2.5 指定从ck自动重启策略
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                3, Time.days(1L),Time.minutes(1L)
//        ));
//        //2.6 设置状态后端
//        env.setStateBackend(new HashMapStateBackend());//本地状态位置
//        env.getCheckpointConfig().setCheckpointStorage(
//                "hdfs://hadoop102:8020/flinkCDC/220828"
//        );//checkpoint状态位置
//        //2.7 设置访问HDFS的用户名
//        System.setProperty("HADOOP_USER_NAME","atguigu");


        //todo 2.读取kafka topic_db主题数据创建主流
        DataStreamSource<String> kafkaDS = env.addSource(KafkaUtil.getFlinkKafkaConsumer("topic_db", "dim_app_220828"));

        //todo 3.过滤并转换为json格式（将数据转换为json对象，并简单过滤）主流
        // 过滤掉非json数据&保留新增、变化以及初始化数据（bootstrap-start、complete和delete数据不要）
        //方法一：
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                //上面已经对null值处理了，如果为null，则返回null，返回值是void，即没有返回，就不需要用out进行输出
                if (value != null) {
                    try {
                        JSONObject jsonObject = JSON.parseObject(value);
                        out.collect(jsonObject);//合法的json就用正常输出
                        //当value不能转化为jsonObject时，就说明value不是合法的json数据，就需要抛异常，我们得try catch
                        //选择上面一行数据：ctrl+alt+t就可以try catch了
                    } catch (JSONException e) {
                        e.printStackTrace();
                        System.out.println("非Json格式数据：" + value);
                        //如果要对非json数据处理的话，就需要用到侧输出流，就不能用flatmap了，要用process
                    }
                }
            }
        });

        //方法二：
//               OutputTag<String> dirtyTag = new OutputTag<String>("Dirty") {
//        };//该类的对象编译的时候会被泛型擦除，但是匿名实现类继承了该类，泛型就定死了为String,就不会有泛型擦除
//        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.process(new ProcessFunction<String, JSONObject>() {
//            @Override
//            public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
//                if (value != null) {
//                    //尝试将数据转换为JSON对象
//                    try {
//                        JSONObject jsonObject = JSON.parseObject(value);
//                        out.collect(jsonObject);
//                    } catch (JSONException e) {
//                        ctx.output(dirtyTag, value);
//                    }
//                }
//            }
//        });




        //todo 4.使用flink CDC 读取mysql配置信息表创建配置流
        MySqlSource<String> mysqlSource = MySqlSource.<String>builder()//注意！！！ 这里的泛型不能丢
                .hostname("hadoop102")
                .port(3306)
                .username("root")
                .password("123456")
                .databaseList("gmall-220828-config")
                .tableList()//可以不指定表，不指定表就默认读取上述数据库的所有表
                .startupOptions(StartupOptions.initial())//初始的一些配置信息也要
                .deserializer(new JsonDebeziumDeserializationSchema())//读取的数据为jsonString类型
                .build();
        
        //读取配置信息表并转化为配置流
        DataStreamSource<String> mysqlDS = env.fromSource(
                mysqlSource,
                WatermarkStrategy.noWatermarks(),
                "MysqlSource");


        //todo 5.将配置信息流处理为广播流
        MapStateDescriptor<String, TableProcess> mapStateDescriptor = new MapStateDescriptor<String, TableProcess>("mapState",String.class,TableProcess.class);
        //看一下下面广播流的返回值类型，并不是<String, TableProcess>，我们需要获取广播状态，把<String, TableProcess>设置进广播状态
        BroadcastStream<String> broadcastDS = mysqlDS.broadcast(mapStateDescriptor);


        //todo 6.连接主流与广播流
        BroadcastConnectedStream<JSONObject, String> connectedStream = jsonObjDS.connect(broadcastDS);

        //todo 7.处理连接流，根据配置信息（广播流）处理(过滤)主流数据
        SingleOutputStreamOperator<JSONObject> hbaseDS = connectedStream.process(new DimTableProcessFuntion(mapStateDescriptor));


        //todo 8.将主流过滤后的维表数据（sink_type为dim的数据）内容写出到phoenix的表内容里
        hbaseDS.print("hbaseDS:主流过滤后的数据要写到phoenix的表内容里>>>>>>>>>>>>");
//        hbaseDS.addSink(JdbcSink.sink("",))//jdbcSink用不了，因为占位符不知道有几个，这个只适合单表查询
        hbaseDS.addSink(new DimSinkFunction());



        //todo 9.启动任务
        env.execute("dimapp");
    }
}
